package com.shujia.flink.core

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import java.lang

object Demo7Process {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    /**
     * process: 一行一行处理数据，可以返回多行，相当于flatmap
     * process:可以用于代替，map,filter,flatmap
     *
     */
    //在DataStream上使用process函数
    val procesDS: DataStream[(String, Int)] = linesDS
      .process(new ProcessFunction[String, (String, Int)] {


        /**
         * 也课题重写open和close方法，可以将初始化的代码放在open中
         *
         */
        override def open(parameters: Configuration): Unit = {
        }

        override def close(): Unit = {
        }

        /**
         * processElement： 将ds的中的数据一条一条传递给processElement，
         * processElement可以输出多条数据，相当于flatmap
         *
         * @param value : 一行数据
         * @param ctx   ： 上下问对象,可以获取到flink的时间
         * @param out   ： 用于见数据发送到下游
         */
        override def processElement(value: String,
                                    ctx: ProcessFunction[String, (String, Int)]#Context,
                                    out: Collector[(String, Int)]): Unit = {

          val clazz: String = value.split(",")(4)

          //将数据发送到下游
          out.collect((clazz, 1))
        }
      })


    procesDS.print()

    env.execute()


  }

}
